package com.atguigu.flink.sqlfunction;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * Created by Smexy on 2022/12/23
 *
 *  UDAF: 聚合函数，输入多(行)，输出 1行1列
 */
public class Demo4_CustomUDAFFunction
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String  createTableSQL = " CREATE TABLE t1( id string, ts bigint , vc int  ) " +
            "                       WITH (  " +
            "                         'connector' = 'filesystem',  " +
            "                         'path' = 'data/t1',   " +
            "                         'format' = 'csv'    " +
            "                            )      ";


        tableEnv.executeSql(createTableSQL);

        Table t1 = tableEnv.from("t1");

        /*
           使用:  在 Table API 中，表值函数是通过 .joinLateral(...) 或者 .leftOuterJoinLateral(...) 来使用的
                            .joinLateral(...)： inner join 取原表和炸裂后字段的交集
                            .leftOuterJoinLateral(...) ： left join 取原表的全部和 炸裂后字段的交集部分

                在 SQL 里面用 JOIN 或者 以 ON TRUE 为条件的 LEFT JOIN 来配合 LATERAL TABLE(<TableFunction>) 的使用。


           -------------------------
            回忆hive，要使用udtf函数，例如explode，语法:

                select
                    id, col1,col2
                from t1
                lateral view explode(x) tmp as col1,col2

                本质：  lateral view 会被翻译为 join!。
                        UDTF如果希望和其他的列一起查询，必须使用Join，把炸裂的结果和一起查询的字段 join 起来！


         */
        //①创建函数对象
        MyAvg myAvg = new MyAvg();

        //注册： 给函数起个名字，使用名字调用函数
        tableEnv.createTemporaryFunction("a",myAvg);

       t1
            .groupBy($("id"))
            .select($("id"),call("a",$("vc")).as("avgVc"))
            .execute()
            .print();

        //SQL  必须注册名字
       /*tableEnv.sqlQuery("select id ,word,length" +
           //内连接
          // "             from t1 , lateral table( a(id) ) ")
           //"             from t1  inner join lateral table( a(id) )  on true")
               //左连接
           "             from t1  left join lateral table( a(id) )  on true")
                .execute()
                .print();*/



       //重命名字段
       /* tableEnv.sqlQuery("select id ,b,c" +
            //左连接
            "             from t1  left join lateral table( a(id) ) as tmp( b,c ) on true")
                .execute()
                .print();*/


    }

    /*
    特点:
       下面几个方法是每个 AggregateFunction 必须要实现的：

            createAccumulator()
            accumulate()
            getValue()
     */
    public static class MyAvg extends AggregateFunction<Double,MyAcc>
    {

        //累加  第一个参数是累加器，第二个参数是输入的数据
        public void accumulate(MyAcc acc, Long vc) {
            acc.sum += vc;
            acc.count += 1;
        }


        //返回最终结果
        @Override
        public Double getValue(MyAcc accumulator) {
            return accumulator.sum / accumulator.count;
        }

        @Override
        public MyAcc createAccumulator() {
            return new MyAcc(0,0d);
        }
    }



    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class MyAcc{
        private Integer count;
        private Double sum;
    }
}
